#include "xlcomm/net/event_loop.h"

#include <sys/eventfd.h>
#include <unistd.h>

#include <thread>

#include "xlcomm/base/logger.h"
#include "xlcomm/net/detail/channel.h"
#include "xlcomm/net/detail/sockets_ops.h"
#include "xlcomm/net/detail/timer_queue.h"
#include "xlcomm/net/poller/poller.h"

namespace xlcomm {
namespace net {

thread_local EventLoop* t_loop_in_this_thread = nullptr;

const int kPollTimeoutMs = 10 * 1000;

// 创建eventfd文件描述符
int CreateEventfd() {
  int fd = ::eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
  if (fd < 0) {
    LOG_SYSFATAL(eventfd);
  }
  return fd;
}

EventLoop::EventLoop()
    : creator_thread_id_(CurrentThread::GetThreadId()),
      looping_(false),
      calling_pending_functors_(false),
      poll_count_(0),
      poll_return_time_(0),
      poller_(Poller::CreatePoller(this)),
      timer_queue_(new TimerQueue(this)),
      wakeup_fd_(CreateEventfd()),
      wakeup_channel_(new Channel(this, wakeup_fd_)) {
  XLCOMM_ASSERT(poller_ != nullptr);
  XLCOMM_ASSERT(wakeup_channel_ != nullptr);
  LOG_INFO << "EventLoop " << this << " created in thread "
           << creator_thread_id_ << ", wakeup_fd=" << wakeup_fd_;
  if (t_loop_in_this_thread == nullptr) {
    t_loop_in_this_thread = this;
  } else {
    LOG_FATAL << "Another EventLoop " << t_loop_in_this_thread
              << " exists in this thread " << creator_thread_id_;
  }
  wakeup_channel_->set_read_callback(std::bind(&EventLoop::HandleRead, this));
  wakeup_channel_->EnableReading();
}

EventLoop::~EventLoop() {
  LOG_INFO << "EventLoop " << this << " of thread " << creator_thread_id_
           << " destructs in thread " << CurrentThread::GetThreadId();
  wakeup_channel_->DisableAll();  // XLCOMM_ASSERT(IsInLoopThread());
  wakeup_channel_->Remove();
  sockets::close(wakeup_fd_);
  t_loop_in_this_thread = nullptr;
}

void EventLoop::Loop() {
  XLCOMM_ASSERT(!looping_);
  XLCOMM_ASSERT(IsInLoopThread());
  LOG_INFO << "EventLoop " << this << " start looping...";
  looping_ = true;
  while (looping_) {
    stop_latch_.SetAndNotifyOnce();
    active_channels_.clear();
    poll_return_time_ = poller_->Poll(kPollTimeoutMs, &active_channels_);
    ++poll_count_;
    if (Logger::GetLogLevel() <= Logger::kTrace) PrintActiveChannels();
    for (Channel* channel : active_channels_) {
      channel->HandleEvent(poll_return_time_);
    }
    DoPendingFunctors();
  }
  LOG_INFO << "EventLoop " << this << " stop looping.";
}

void EventLoop::RunInLoop(Functor func) {
  if (IsInLoopThread()) {
    func();
  } else {
    QueueInLoop(std::move(func));
  }
}

void EventLoop::QueueInLoop(Functor func) {
  {
    // pending_functors_暴露给了其他线程，因此需要加锁
    std::lock_guard<std::mutex> guard(functor_mutex_);
    pending_functors_.push_back(std::move(func));
  }
  if (!IsInLoopThread() || calling_pending_functors_) {
    Wakeup();
  }
}

TimerId EventLoop::RunAt(Timestamp time, TimerCallback cb) {
  return timer_queue_->AddTimer(std::move(cb), time, 0.0);
}

TimerId EventLoop::RunAfter(double delay_second, TimerCallback cb) {
  Timestamp now = DateTime::NowTimestamp();
  return timer_queue_->AddTimer(std::move(cb),
                                DateTime::AddSeconds(now, delay_second), 0.0);
}

TimerId EventLoop::RunEvery(double interval_second, TimerCallback cb) {
  Timestamp now = DateTime::NowTimestamp();
  return timer_queue_->AddTimer(std::move(cb),
                                DateTime::AddSeconds(now, interval_second),
                                interval_second);
}

void EventLoop::DoPendingFunctors() {
  calling_pending_functors_ = true;
  std::vector<Functor> functors;
  {
    // 一方面减小临界区长度，一方面避免回调函数再调QueueInLoop死锁
    std::lock_guard<std::mutex> guard(functor_mutex_);
    functors.swap(pending_functors_);
  }
  for (const Functor& func : functors) {
    // 如果在IO线程调QueueInLoop，而此时正在调pending_functors，也必须Wakeup
    func();
  }
  calling_pending_functors_ = false;
}

void EventLoop::Quit() {
  stop_latch_.Wait();  // 防止先调Quit()后调Loop()导致进入不了监听
  LOG_INFO << "quit EventLoop " << this;
  looping_ = false;
  if (!IsInLoopThread()) {
    Wakeup();
  }
}

void EventLoop::Wakeup() {
  uint64_t one = 1;  // eventfd只能读写8-byte integer value
  ssize_t n = sockets::write(wakeup_fd_, &one, sizeof(one));
  if (n != sizeof(one)) {
    LOG_SYSERROR(write) << "EventLoop writes " << n << " bytes instead of 8!";
  }
}

void EventLoop::HandleRead() {
  LOG_TRACE << "EventLoop(wakeup_fd:" << wakeup_fd_ << ") read some";
  uint64_t one = 0;
  ssize_t n = sockets::read(wakeup_fd_, &one, sizeof(one));
  if (n != sizeof(one)) {
    LOG_SYSERROR(write) << "EventLoop reads " << n << " bytes instead of 8!";
  }
}

void EventLoop::UpdateChannel(Channel* channel) {
  XLCOMM_ASSERT(channel->loop() == this);
  XLCOMM_ASSERT(IsInLoopThread());
  poller_->UpdateChannel(channel);
}

void EventLoop::RemoveChannel(Channel* channel) {
  XLCOMM_ASSERT(channel->loop() == this);
  XLCOMM_ASSERT(IsInLoopThread());
  poller_->RemoveChannel(channel);
}

bool EventLoop::HasChannel(Channel* channel) const {
  XLCOMM_ASSERT(channel->loop() == this);
  XLCOMM_ASSERT(IsInLoopThread());
  return poller_->HasChannel(channel);
}

EventLoop* EventLoop::GetEventLoopOfCurrentThread() {
  return t_loop_in_this_thread;
}

void EventLoop::PrintActiveChannels() const {
  if (active_channels_.size() > 0)
    LOG_TRACE << "active channels's triggered events as below:";
  for (const Channel* channel : active_channels_) {
    LOG_TRACE << "fd:" << channel->fd() << "{"
              << channel->TriggeredEventsToString() << "} ";
  }
}

}  // namespace net
}  // namespace xlcomm